package pipeline;

import com.mysql.jdbc.exceptions.jdbc4.MySQLIntegrityConstraintViolationException;
import model.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import us.codecraft.webmagic.ResultItems;
import us.codecraft.webmagic.Task;
import us.codecraft.webmagic.pipeline.Pipeline;

import java.sql.*;
import java.util.ArrayList;
import java.util.Date;
import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import util.Config;
import util.JSONUtil;

/**
 * @author J0101
 * @Description 使用Kafka插入数据库的不同数据库中
 */
public class KafkalPipeline implements Pipeline{
    private static final String HOST = Config.getCrawler_KAFKA_SERVER();
    private static final String PREFIX = Config.getCrawler_PREFIX();

    private Logger logger = LoggerFactory.getLogger(getClass());
    private static Producer<String, String> producer;

    public KafkalPipeline(){
        Properties props = new Properties();
        props.put("bootstrap.servers", HOST);
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        producer = new KafkaProducer<>(props);
    }
    @Override
    public void process(ResultItems resultItems, Task task) {
        String job_id = resultItems.getRequest().getExtra("job_id").toString();
        BaseRegisterInfo baseRegisterInfo = (BaseRegisterInfo)resultItems.get("BaseRegisterInfo");
        if(baseRegisterInfo != null){
            insertBaseRegisterInfo(baseRegisterInfo,job_id);
        }
        BaseBusinessInfo businessInfo = (BaseBusinessInfo)resultItems.get("BaseBusinessInfo");
        if(businessInfo != null){
            insertBaseBusinessInfo(businessInfo,job_id);
        }
        ArrayList<BaseMainStaff> baseMainStaffs = resultItems.get("BaseMainStaff");
        if(baseMainStaffs != null){
            insertBaseMainStaff(baseMainStaffs,job_id);
        }
        ArrayList<BaseShareholderInfo> baseShareholderInfos = resultItems.get("BaseShareholderInfo");
        if(baseShareholderInfos != null){
                insertBaseShareholderInfo(baseShareholderInfos,job_id);

        }
        ArrayList<BaseInvestment> baseInvestments = resultItems.get("BaseInvestment");
        if(baseInvestments != null){
                insertBaseInvestment(baseInvestments,job_id);

        }

        ArrayList<BaseChange> baseChanges = resultItems.get("BaseChange");
        if(baseChanges != null){
            insertBaseChange(baseChanges,job_id);

        }
        BaseProduct baseProduct = (BaseProduct) resultItems.get("BaseProduct");
        if(baseProduct != null){
            insertBaseProduct(baseProduct,job_id);
        }
        ArrayList<BaseProductList> baseProductLists = resultItems.get("BaseProductList");
        if(baseProductLists != null){
            insertBaseProductList(baseProductLists,job_id);
        }
        ArrayList<EquityPledge> equityPledges = resultItems.get("EquityPledge");
        if(equityPledges != null){
            insertEquityPledge(equityPledges,job_id);

        }
        FengxianTag fengxianTag = (FengxianTag) resultItems.get("FengxianTag");
        if(fengxianTag != null){
                insertFengxianTag(fengxianTag,job_id);

        }
        SusongTag susongTag = (SusongTag) resultItems.get("SusongTag") ;
        if(susongTag != null){
                insertSusongTag(susongTag,job_id);
        }
//        System.out.println(resultItems.get("")+"插入完成");
    }
    public void insertBaseRegisterInfo(BaseRegisterInfo baseRegisterInfo,String job_id){


    }
    public void insertBaseBusinessInfo(BaseBusinessInfo businessInfo,String job_id){
        businessInfo.setJob_id(job_id);
        businessInfo.setCro_id(businessInfo.getCredit_code()+"_"
                                +businessInfo.getRegistration_number()+"_"
                                +businessInfo.getOrganization_code());
        String body = null;
        try {
            body = String.valueOf(JSONUtil.objectToJson(businessInfo));
        } catch (Exception e) {
            e.printStackTrace();
        }
//        System.out.println("Send->[" + body + "]");
        logger.debug(body);
        producer.send(new ProducerRecord<String, String>(PREFIX+"."+"crawler_"+"BaseBusinessInfo", String.valueOf(new Date().getTime()), body));

    }
    public void insertBaseProduct(BaseProduct baseProduct,String job_id){


    }
    public void insertBaseProductList(ArrayList<BaseProductList> baseProductLists,String job_id ) {


    }
    public void insertBaseMainStaff(ArrayList<BaseMainStaff> baseMainStaffs,String job_id ) {
        for (BaseMainStaff baseMainStaff:baseMainStaffs){
            String body = null;
            baseMainStaff.setJob_id(job_id);
            try {
                body = String.valueOf(JSONUtil.objectToJson(baseMainStaff));
            } catch (Exception e) {
                e.printStackTrace();
            }
            producer.send(new ProducerRecord<String, String>(PREFIX+"."+"crawler_"+"BaseMainStaff", String.valueOf(new Date().getTime()), body));
        }

    }
    public void insertBaseShareholderInfo(ArrayList<BaseShareholderInfo> baseShareholderInfos,String job_id ){

        for (BaseShareholderInfo baseShareholderInfo:baseShareholderInfos){
            String body = null;
            baseShareholderInfo.setJob_id(job_id);
            try {
                body = String.valueOf(JSONUtil.objectToJson(baseShareholderInfo));
            } catch (Exception e) {
                e.printStackTrace();
            }
            producer.send(new ProducerRecord<String, String>(PREFIX+"."+"crawler_"+"BaseShareholderInfo", String.valueOf(new Date().getTime()), body));
        }
    }
    public void insertEquityPledge(ArrayList<EquityPledge> equityPledges,String job_id ){

    }
    public void insertBaseChange(ArrayList<BaseChange> baseChanges ,String job_id){


    }
    public void insertBaseInvestment(ArrayList<BaseInvestment> baseInvestments ,String job_id){
        for (BaseInvestment baseInvestment:baseInvestments){
            String body = null;
            baseInvestment.setJob_id(job_id);
            try {
                body = String.valueOf(JSONUtil.objectToJson(baseInvestment));
            } catch (Exception e) {
                e.printStackTrace();
            }
            producer.send(new ProducerRecord<String, String>(PREFIX+"."+"crawler_"+"BaseInvestment", String.valueOf(new Date().getTime()), body));
        }

    }
    public void insertFengxianTag(FengxianTag fengxianTag ,String job_id){


    }
    public void insertSusongTag(SusongTag susongTag ,String job_id) {

    }

}
